package com.yanfei.zero.apis;

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import com.yanfei.zero.application.handler.lock.AddLock;
import com.yanfei.zero.application.handler.log.AddLog;
import com.yanfei.zero.application.handler.secret.IgnoreDecrypt;
import com.yanfei.zero.domain.user.model.User1;
import com.yanfei.zero.domain.user.model.perms.Auth;
import com.yanfei.zero.domain.user.repository.User1Mapper;
import com.yanfei.zero.domain.user.repository.User1Repository;
import com.yanfei.zero.infrastructure.config.DeveloperProperty;

import cn.dev33.satoken.stp.SaTokenInfo;
import cn.dev33.satoken.stp.StpUtil;
import cn.dev33.satoken.util.SaResult;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.NumberUtil;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.transaction.Transactional;
import jakarta.validation.Valid;

@RestController
@Tag(name="用户")
@Auth(id=1000,name="用户")
public class UserController {
	
	@Autowired  
    private User1Mapper user1Mapper;  
	
	private static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
	
	// 超时时间设置为3s，用于演示客户端自动重连
	private static SseEmitter sseEmitter = new SseEmitter(30000L);
    @Autowired
	private DeveloperProperty developerProperty;
	@Autowired
	private User1Repository user1Repository;

	@Operation(summary = "添加100万用户")
    @PostMapping("/users")
    @IgnoreDecrypt
    @Auth(id=1,name="添加用户")
//    @AddLog
    // @RateLimiter(key = "tets")
//    @Encrypt
    Object addUser(@Valid @RequestBody User1 user1) {
//        user1Repository.save(user1);  
        // persisting the user
        IntStream.range(0, 1000000).parallel().forEach(i -> {  
            User1 user = new User1(); 
            long id = IdUtil.getSnowflakeNextId();
            user.setId(id);
            user.setName("User" + id);  
            user.setEmail("user" +id  + "@example.com");  
            user.setMoney(user1.getMoney());  
            user.setPassword("pwd"+id);
            user1Repository.save(user);  
        });  
        return user1;
    }
	
	@Operation(summary = "流式查询")
    @PostMapping("/streamQuery")
	@AddLock("lock:121")
	@IgnoreDecrypt
    @Transactional
    @AddLog
    Object getUser(@RequestBody User1 user1) {
		
		long b = System.currentTimeMillis();
		user1Mapper.streamQuery(r->{
			User1 u1 = r.getResultObject();
			// do business
			System.out.println(u1.getId());
			
			});
		System.out.println(System.currentTimeMillis()-b);
    	return "success";
    }
	
	
	@Operation(summary = "分页查询")
    @PostMapping("/pageQuery")
	@AddLock("lock:121")
	@IgnoreDecrypt
    @Transactional
    @AddLog
    Object getpageUser(@RequestBody User1 user1) {
		
		
        int pageSize = 1000; // 每页数据条数
        AtomicInteger pageNum = new AtomicInteger(1); // 使用 AtomicInteger 保证原子性操作
        long startTime = System.currentTimeMillis();

        // 创建一个线程池
        ExecutorService executor = Executors.newFixedThreadPool(5); // 这里使用了固定大小为 5 的线程池

        ConcurrentLinkedQueue<User1> allUsers = new ConcurrentLinkedQueue<>();
        while (true) {
            Map<String, Object> parameterMap = new HashMap<>();
            parameterMap.put("pageSize", pageSize);
            parameterMap.put("offset", (pageNum.getAndIncrement() - 1) * pageSize); // 使用 getAndIncrement() 方法获取递增后的值

            // 提交分段查询任务到线程池
            Future<List<User1>> future = executor.submit(()->{
            	List<User1> us = user1Mapper.selectAll(parameterMap);
            	return us;
            });

            try {
                List<User1> userList = future.get(); // 获取查询结果
                allUsers.addAll(userList); // 将查询结果添加到总结果集
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }

            if (pageNum.get() > 10) { // 假设只查询前 10 页数据
                break;
            }
        }

        // 关闭线程池
        executor.shutdown();

        // 处理所有查询结果
        allUsers.parallelStream().forEach(s -> System.out.println(s.getId()));

        long endTime = System.currentTimeMillis();
        System.out.println("Total time: " + (endTime - startTime) + " ms");
    	return "success";
    }
    
    // standard constructors / other methods


	@Operation(summary = "登录")
    @PostMapping("/login")
	 @AddLock("lock:1")
    @AddLog
	@IgnoreDecrypt
    Object login(@RequestBody User1 user1) {
		
		
        // persisting the user
		Optional<User1> op = user1Repository.findByNameAndPassword(user1.getName(),user1.getPassword());
		User1 user= op.get(); // auto throw exception

    	user1Repository.save(user);
    	// 登录
    	StpUtil.login(user.getId());
        // 获取用户信息
		SaTokenInfo tokenInfo = StpUtil.getTokenInfo();
		
    	return SaResult.data(tokenInfo);
    }
}
